package com.ylz.sjzx.sio.core.group.manager;


import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alipay.remoting.Connection;
import com.alipay.remoting.exception.RemotingException;
import com.alipay.remoting.rpc.RpcClient;
import com.alipay.remoting.rpc.RpcServer;
import com.alipay.remoting.rpc.protocol.UserProcessor;
import com.ylz.sjzx.sio.common.core.constant.Const;
import com.ylz.sjzx.sio.common.core.utils.HostUtil;
import com.ylz.sjzx.sio.core.group.entity.RpcClientEntity;
import com.ylz.sjzx.sio.core.group.entity.RpcServiceEntity;
import com.ylz.sjzx.sio.core.group.model.GroupRegister;
import com.ylz.sjzx.sio.core.group.model.RpcProperties;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 远程服务统一管理器
 *
 * @author
 * @version 1.0
 * @date 2021/1/6
 */
@Component
@Slf4j
public class GroupRpc {
    /**
     * 本地rpcServer
     */
    public static RpcServer localRpcServer;
    /**
     * 本地rpcClient
     */
    public static RpcClient localRpcClient;
    /**
     * client connect管理器
     * connect 为空
     */
    public static Map<String, GroupRegister> serviceAddr = new ConcurrentHashMap<>(10);
    public static RpcProperties localRemotingProperties;
    /**
     *
     */
    private static ExecutorService senMessageService = Executors.newFixedThreadPool(4 + 1);
    /**
     * 端口
     */
    private Integer servicePort;
    /**
     * 组名称
     */
    private String groupName;
    /**
     * 组角色
     */
    private String groupRole;
    /**
     * 处理消息接收的
     */
    private List<UserProcessor> rpcUserProcessors;

    public GroupRpc(@Autowired List<UserProcessor> rpcUserProcessors,
                    @Value("${im.group.service.port:1234}") Integer servicePort,
                    @Value("${im.group.service.name:default}") String groupServiceName,
                    @Value("${im.group.service.role:worker}") String groupRole) {
        this.rpcUserProcessors = rpcUserProcessors;
        this.servicePort = servicePort;
        this.groupName = groupServiceName;

        this.groupRole = groupRole;
        //初始化本地rpcService
        localRpcServer = new RpcServiceEntity(this.createLocalProperties()).getRpcServer();
        //初始化本地rpcClient
        localRpcClient = new RpcClientEntity().getRpcClient();
    }

    public static void broadcastMessage(Object message) {
        senMessageService.execute(new Runnable() {
            @Override
            public void run() {
                GroupRpc.serviceAddr.forEach((k, v) -> {
                    sendOneWay(k, message);
                });
            }
        });
    }

    public static void sendMessage(String str, Object message) {
        if (!serviceAddr.containsKey(str)) {
            return;
        }
        sendOneWay(str, message);
    }

    @SneakyThrows
    private static void sendOneWay(String str, Object message) {
        localRpcClient.oneway(str, message);
    }

    /**
     * 获取本机的properties
     *
     * @return
     */
    public synchronized RpcProperties createLocalProperties() {
        if (localRemotingProperties == null) {
            localRemotingProperties = new RpcProperties();
            localRemotingProperties.setIp(HostUtil.getHostAddress());
            localRemotingProperties.setPort(servicePort);
            localRemotingProperties.setGroupName(this.groupName);
            localRemotingProperties.setGroupRole(this.groupRole);
            localRemotingProperties.setRpcUserProcessors(rpcUserProcessors);
        }
        return localRemotingProperties;
    }

    /**
     * 创建client并加入管理
     */
    public synchronized void addRpcService(Instance instance) {
        RpcProperties properties = new RpcProperties();
        properties.setIp(instance.getIp());
        properties.setPort(instance.getPort());
        //创建addrString
        String addr = properties.getAddr(instance.getIp(), instance.getPort());
        //本机不注册
        if (properties.ifLocalService(addr)) {
            log.debug("====address={},本机不注册=====", addr);
            return;
        }
        //判断是否已经注册
        if (serviceAddr.containsKey(addr)) {
            log.debug("====address={},已注册过=====", addr);
            return;
        }
        Connection standaloneConnection;
        try {
            standaloneConnection = localRpcClient.createStandaloneConnection(addr, 2000);
        } catch (RemotingException e) {
            log.error("====创建远程路由超时======", e);
            return;
        } catch (Exception e) {
            log.error("====创建远程路错误======", e);
            return;
        }

        GroupRegister groupRegister = GroupRegister.builder().groupName(instance.getMetadata().get(Const.GROUP_NAME))
                .gruopRole(instance.getMetadata().get(Const.GROUP_ROLE))
                .ip(instance.getIp())
                .port(instance.getPort())
                .connection(standaloneConnection).build();
        serviceAddr.put(addr, groupRegister);
        log.info("====address={},注册成功=====", addr);
    }

    /**
     * 删除client
     *
     * @param ip
     * @param port
     */
    public synchronized void removeRpcService(String ip, int port) {
        RpcProperties properties = new RpcProperties();
        properties.setIp(ip);
        properties.setPort(port);
        //创建addrString
        String addr = properties.getAddr(ip, port);
        serviceAddr.remove(addr);
        //同时清除连接
        localRpcClient.closeConnection(addr);
        log.info("===不健康服务{}清除======", addr);
    }

}
